/*
 * To change this license header, choose License Headers in Project Properties.
 * To change this template file, choose Tools | Templates
 * and open the template in the editor.
 */
package cn.ac.iie.di.ban.handler;

import cn.ac.iie.di.ban.commons.RuntimeEnv;
import cn.ac.iie.di.ban.tools.DataBuffer;
import cn.ac.iie.di.ban.tools.HttpPostHelper;
import cn.ac.iie.di.ban.tools.InfluxDBHelper;
import cn.ac.iie.di.ban.tools.TimeControler;
import java.rmi.server.UID;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import org.apache.log4j.Logger;

/**
 *
 * @author xiaoliang
 */
public class SE2DBWorkerInfluxDB extends SE2DBWorker {

    private static org.apache.log4j.Logger logger = Logger.getLogger(SE2DBWorkerInfluxDB.class.getName());
    private InfluxDBHelper recNumDB;
    private InfluxDBHelper dataSizeDB;
    private InfluxDBHelper flowRate;

//    private HttpPostHelper metaFind;
    private TimeControler timeControler;

    public SE2DBWorkerInfluxDB(HttpSe2DBHandler httpSe2DBHandler, String pDWType) {
        super(httpSe2DBHandler, pDWType);
    }

    @Override
    public void init() throws Exception {
        recNumDB = InfluxDBHelper.getInfluxDBHelper(load2DBEntity.getDb(), load2DBEntity.getRecNumTable(), load2DBEntity.getIp());
        dataSizeDB = InfluxDBHelper.getInfluxDBHelper(load2DBEntity.getDb(), load2DBEntity.getDataSizeTable(), load2DBEntity.getIp());
        flowRate = InfluxDBHelper.getInfluxDBHelper(load2DBEntity.getDb(), load2DBEntity.getFlowRate(), load2DBEntity.getIp());
//        metaFind = new HttpPostHelper(RuntimeEnv.getParamAsString(RuntimeEnv.METADATAURL));
        timeControler = new TimeControler(load2DBEntity);
        Thread thread = new Thread(timeControler);
        thread.start();
        logger.info("read insert in to db is start");
    }

    @Override
    public void loadData0() {
        try {
            if (docRecordSet.isEmpty()) {
                return;
            }
            start_time = System.currentTimeMillis();
            List<Map<String, Object>> readyclList = new ArrayList();
            List<Map<String, String>> docRecords = new ArrayList();
            List<Map<String, String>> flowRateList = new ArrayList();
            for (Map<String, Object> record : docRecordSet) {
                String scope = record.get("scope").toString();
                if ("2,3".contains(scope)) {
                    readyclList.add(record);
                    continue;
                } else if ("-1".equals(scope)) {
                    initFlowMsg(flowRateList, record);
                    continue;
                }
                List<Map<String, String>> yclrecords = yclMsg(record);
                docRecords.addAll(yclrecords);
            }
            if (!readyclList.isEmpty()) {
                while (!timeControler.addMsg(readyclList)) {
                    try {
                        logger.info("timeControle is working waitting ...");
                        Thread.sleep(200);
                    } catch (Exception e) {
                    }
                }
            }
            List<Map<String, String>> recNumMsgs = new ArrayList();
            List<Map<String, String>> dataSizeMsgs = new ArrayList();
            if (!docRecords.isEmpty()) {
                shuntmsg(docRecords, recNumMsgs, dataSizeMsgs);
                recNumDB.batchInsert(recNumMsgs, "rec_num");
                dataSizeDB.batchInsert(dataSizeMsgs, "data_size");
            }
            logger.debug("insert into influx " + (end_time - start_time));
            end_time = System.currentTimeMillis();
            if (!flowRateList.isEmpty()) {
                flowRate.batchInsert(flowRateList, "value");
                logger.info("receive flow_rate " + flowRateList.size() + " |expect db " + flowRateList.size() + " record to " + load2DBEntity.getFlowRate() + " successfully using " + (end_time - start_time) + " ms ");
            }
            end_time = System.currentTimeMillis();
            logger.info("save req " + docRecords.size() + "|expect task" + (docRecordSet.size() - flowRateList.size()) + " record to " + load2DBEntity.getDataSizeTable() + load2DBEntity.getRecNumTable() + " successfully " + " using " + (end_time - start_time) + " ms ");
            docRecords.clear();
            docRecordSet.clear();
        } catch (Exception e) {
            recNumDB.connect();
            dataSizeDB.connect();
            flowRate.connect();
            logger.error(e.getMessage(), e);
        }
    }

    public void initFlowMsg(List<Map<String, String>> flowRateList, Map<String, Object> record) {
        Map<String, Object> flowMap = new HashMap<>();
        Map<String, String> flowMapStr = new HashMap<>();
        String uid = record.get("uid").toString();
        flowMap.putAll(record);
        flowMap.putAll(DataBuffer.getAgencyAllbyUid(uid));
        flowMap.remove("scope");
        for(String str : flowMap.keySet()){
            flowMapStr.put(str, flowMap.get(str).toString().trim());
        }
        flowRateList.add(flowMapStr);
    }
    
    @Override
    public void freeSource() {

    }
}
